/*
 * Copyright 2010-2020 Alfresco Software, Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.activiti.engine.impl.bpmn.behavior;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.activiti.bpmn.model.Activity;
import org.activiti.bpmn.model.FlowElement;
import org.activiti.bpmn.model.FlowNode;
import org.activiti.bpmn.model.ParallelGateway;
import org.activiti.engine.ActivitiException;
import org.activiti.engine.ActivitiOptimisticLockingException;
import org.activiti.engine.debug.Contants;
import org.activiti.engine.delegate.DelegateExecution;
import org.activiti.engine.impl.context.Context;
import org.activiti.engine.impl.interceptor.CommandContext;
import org.activiti.engine.impl.interceptor.Session;
import org.activiti.engine.impl.persistence.entity.ExecutionEntity;
import org.activiti.engine.impl.persistence.entity.ExecutionEntityImpl;
import org.activiti.engine.impl.persistence.entity.ExecutionEntityManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ziroom.common.api.constants.Constant;

/**
 * Implementation of the Parallel Gateway/AND gateway as defined in the BPMN 2.0 specification.
 *
 * The Parallel Gateway can be used for splitting a path of execution into multiple paths of executions (AND-split/fork behavior), one for every outgoing sequence flow.
 *
 * The Parallel Gateway can also be used for merging or joining paths of execution (AND-join). In this case, on every incoming sequence flow an execution needs to arrive, before leaving the Parallel
 * Gateway (and potentially then doing the fork behavior in case of multiple outgoing sequence flow).
 *
 * Note that there is a slight difference to spec (p. 436): "The parallel gateway is activated if there is at least one Token on each incoming sequence flow." We only check the number of incoming
 * tokens to the number of sequenceflow. So if two tokens would arrive through the same sequence flow, our implementation would activate the gateway.
 *
 * Note that a Parallel Gateway having one incoming and multiple outgoing sequence flow, is the same as having multiple outgoing sequence flow on a given activity. However, a parallel gateway does NOT
 * check conditions on the outgoing sequence flow.
 *
 */
public class ParallelGatewayActivityBehavior extends GatewayActivityBehavior {

	private static final long serialVersionUID = 1840892471343975524L;

	private static Logger logger = LoggerFactory.getLogger(ParallelGatewayActivityBehavior.class);

	@Override
	public void execute(DelegateExecution execution) {

	    // First off all, deactivate the execution
	    execution.inactivate();

	    // Join
	    FlowElement flowElement = execution.getCurrentFlowElement();
	    ParallelGateway parallelGateway = null;
	    if (flowElement instanceof ParallelGateway) {
	    	parallelGateway = (ParallelGateway) flowElement;
	    } else {
	    	throw new ActivitiException("Programmatic error: parallel gateway behaviour can only be applied to a ParallelGateway instance, but got an instance of " + flowElement);
	    }

	    // lockFirstParentScope(execution);

	    Collection<ExecutionEntity> joinedExecutions = null;
	    CommandContext context = Context.getCommandContext();
	    ExecutionEntityManager executionEntityManager = context.getExecutionEntityManager();
	    int nbrOfExecutionsToJoin = parallelGateway.getIncomingFlows().size();
	    boolean isMainThread = false; // 是否主线程
	    boolean isEnter = true; // 是否是入口网关
	    if(nbrOfExecutionsToJoin > 1){
	    	isEnter = false;
	    	isMainThread = mainThreadDeal(execution);
	    	// 只针对处理合并的并行网关需要提前刷新
	    	forceFlushDb();
	    	// 查询已经执行完的并行分支(查询结果还会入缓存)
	    	joinedExecutions = executionEntityManager.findDbInactiveExecutionsByActivityIdAndProcessInstanceId(execution.getCurrentActivityId(), execution.getProcessInstanceId());
	    }else{
	    	DelegateExecution multiInstanceExecution = null;
	    	if (hasMultiInstanceParent(parallelGateway)) {
	    		multiInstanceExecution = findMultiInstanceParentExecution(execution);
	    	}
	    	joinedExecutions = executionEntityManager.findInactiveExecutionsByActivityIdAndProcessInstanceId(execution.getCurrentActivityId(), execution.getProcessInstanceId());
	    	if (multiInstanceExecution != null) {
	    		joinedExecutions = cleanJoinedExecutions(joinedExecutions, multiInstanceExecution);
	    	}
	    }

	    int nbrOfExecutionsCurrentlyJoined = joinedExecutions.size();

	    // Fork
	    // Is needed to set the endTime for all historic activity joins
	    // 记录并行流程结束
	    context.getHistoryManager().recordActivityEnd((ExecutionEntity) execution, null);

	    if ((!isEnter && isMainThread) || isEnter) {
	    	// 不是入口网关，且是主线程
	    	// 是入口网关
	    	// Fork
	    	if (logger.isDebugEnabled()) {
	    		logger.debug("parallel gateway '{}' activates: {} of {} joined", execution.getCurrentActivityId(), nbrOfExecutionsCurrentlyJoined, nbrOfExecutionsToJoin);
	    	}

	    	if (parallelGateway.getIncomingFlows().size() > 1) {
		        // All (now inactive) children are deleted.
		        for (ExecutionEntity joinedExecution : joinedExecutions) {
		        	// 当前的execution会被复用，所以不会被删除，但是会加上完成时间
		        	if (!joinedExecution.getId().equals(execution.getId())) {
		        		executionEntityManager.deleteExecutionAndRelatedData(joinedExecution, null);
		        	} else {
		        		context.getHistoryManager().recordActivityEnd(joinedExecution, null);
		        	}
		        }
		        forceFlushDb();
		        // 这里把execution重新刷入缓存，用于更新操作
		        forceToCache(execution);
	    	}

	    	// TODO: potential optimization here: reuse more then 1 execution, only 1 currently
	    	Context.getAgenda().planTakeOutgoingSequenceFlowsOperation((ExecutionEntity) execution, false); // false -> ignoring conditions on parallel gw

	    } else {
	    	if (logger.isDebugEnabled()) {
		    	logger.debug("parallel gateway '{}' does not activate: {} of {} joined", execution.getCurrentActivityId(), nbrOfExecutionsCurrentlyJoined, nbrOfExecutionsToJoin);
		    }
	    }

	}

	/**
	 * 主线程处理
	 */
	@SuppressWarnings("unchecked")
	private boolean mainThreadDeal(DelegateExecution execution){
		// 判断自己是不是主线程，如果是主线程会阻塞等待求他线程完成，否则直接返回
		List<CompletableFuture<Object>> contextCfList = Context.getPgatewayCompletableFutureList();
    	if(contextCfList != null){
    		Context.removePgatewayCompletableFutureList();
    		for (CompletableFuture<Object> cf : contextCfList) {
    			try {
    				// 理论上，setVariables都是在parent上做设置，那既然各个线程都是设置的同一个parent，这里主线程为什么还要设置一遍？
    				// 因为parent被各个线程共享，设置的时候又没有加锁，所以这里为了可见性就在主线程重新设置了一遍
    				execution.setTransientVariables((Map<String, Object>)cf.get(Constant.DEFAULT_THREAD_TIMEOUT, TimeUnit.SECONDS));
				} catch (InterruptedException e) {
					logger.error(Contants.DEBUG_PREFIX + " interrupted error", e);
				} catch (ExecutionException e) {
					logger.error(Contants.DEBUG_PREFIX + " execution error", e);
				} catch (TimeoutException e) {
					logger.error(Contants.DEBUG_PREFIX + " timeout error", e);
				} catch (Throwable e) {
					logger.error(Contants.DEBUG_PREFIX + " unknow error", e);
				}
			}
    		return true;
    	}
    	return false;
	}

	/**
	 * 强制把数据刷入数据库
	 */
	private void forceFlushDb() {
		CommandContext commandContext = Context.getCommandContext();
		Map<Class<?>, Session> sessions = commandContext.getSessions();
		for (Session session : sessions.values()) {
			// 清空insertedObjects等
			session.flush();
		}
		for (Session session : sessions.values()) {
			// 清空cachedObjects
			session.clear();
		}
	}

	/**
	 * 强制把数据刷入缓存
	 */
	private void forceToCache(DelegateExecution execution) {
		CommandContext commandContext = Context.getCommandContext();
		commandContext.getExecutionEntityManager().update((ExecutionEntityImpl)execution);
	}

	/**
	 * 防止多个任务并行执行
	 *
	 * @return
	 */
	public boolean lockJob(DelegateExecution execution) {
		try {
			CommandContext commandContext = Context.getCommandContext();
			commandContext.getExecutionEntityManager().updateProcessInstanceLockTime(execution.getProcessInstanceId());
			return true;
		} catch (ActivitiOptimisticLockingException e) {
			return false;
		}
	}

	protected Collection<ExecutionEntity> cleanJoinedExecutions(Collection<ExecutionEntity> joinedExecutions, DelegateExecution multiInstanceExecution) {
		List<ExecutionEntity> cleanedExecutions = new ArrayList<ExecutionEntity>();
	    for (ExecutionEntity executionEntity : joinedExecutions) {
	    	if (isChildOfMultiInstanceExecution(executionEntity, multiInstanceExecution)) {
	    		cleanedExecutions.add(executionEntity);
	    	}
	    }
	    return cleanedExecutions;
	}

	protected boolean isChildOfMultiInstanceExecution(DelegateExecution executionEntity, DelegateExecution multiInstanceExecution) {
		boolean isChild = false;
		DelegateExecution parentExecution = executionEntity.getParent();
		if (parentExecution != null) {
			if (parentExecution.getId().equals(multiInstanceExecution.getId())) {
				isChild = true;
			} else {
				boolean isNestedChild = isChildOfMultiInstanceExecution(parentExecution, multiInstanceExecution);
				if (isNestedChild) {
					isChild = true;
				}
			}
		}

		return isChild;
	}

	protected boolean hasMultiInstanceParent(FlowNode flowNode) {
		boolean hasMultiInstanceParent = false;
		if (flowNode.getSubProcess() != null) {
			if (flowNode.getSubProcess().getLoopCharacteristics() != null) {
				hasMultiInstanceParent = true;
			} else {
				boolean hasNestedMultiInstanceParent = hasMultiInstanceParent(flowNode.getSubProcess());
				if (hasNestedMultiInstanceParent) {
					hasMultiInstanceParent = true;
				}
			}
		}

		return hasMultiInstanceParent;
	}

	protected DelegateExecution findMultiInstanceParentExecution(DelegateExecution execution) {
		DelegateExecution multiInstanceExecution = null;
		DelegateExecution parentExecution = execution.getParent();
		if (parentExecution != null && parentExecution.getCurrentFlowElement() != null) {
			FlowElement flowElement = parentExecution.getCurrentFlowElement();
			if (flowElement instanceof Activity) {
				Activity activity = (Activity) flowElement;
				if (activity.getLoopCharacteristics() != null) {
					multiInstanceExecution = parentExecution;
				}
			}

			if (multiInstanceExecution == null) {
				DelegateExecution potentialMultiInstanceExecution = findMultiInstanceParentExecution(parentExecution);
				if (potentialMultiInstanceExecution != null) {
					multiInstanceExecution = potentialMultiInstanceExecution;
				}
			}
		}

		return multiInstanceExecution;
	}

}
